package report

import Configer.Configer
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}
import utils.KPITotal

//网络类型
object NettypeAnalysis {
  def main(args: Array[String]): Unit = {
    //sparkContext
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local[*]")
    conf.set("spark.serializer",Configer.serializer)
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)
    //读取数据
    val dataFrame = sQLContext.read.parquet("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\parquet")
    //处理数据
    val result: RDD[(String, List[Double])] = dataFrame.map(row => {
      val netName = row.getAs[String]("networkmannername")
      val list = KPITotal.KPI(row)
      (netName, list)
    }).reduceByKey {
      (list1, list2) => list1 zip list2 map (li => li._1 + li._2)
    }
    //存储数据
    import sQLContext.implicits._
    result.map(tp=>(tp._1,tp._2(0),tp._2(1),tp._2(2),tp._2(3),tp._2(4),tp._2(7),tp._2(8),tp._2(5),tp._2(6)))
        .toDF("netName","1","2","3","4","5","6","7","8","9")
      .coalesce(1).write.mode(SaveMode.Overwrite).json("E:\\小牛项目\\DMP广告项目34期\\资料PDF\\network")

    sc.stop()

  }
}
